﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

#region

using System.Linq;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;

#endregion

namespace AzureMRCore.DataModel
{
    public class JobDataSource
    {
        private readonly CloudStorageAccount _storageAccount;

        public JobDataSource()
        {
            _storageAccount =
                CloudStorageAccount.Parse(RoleEnvironment.GetConfigurationSettingValue("DataConnectionString"));
        }

        public IQueryable<JobDataModel> GetJobs()
        {
            CloudTableClient tableClient = _storageAccount.CreateCloudTableClient();
            tableClient.CreateTableIfNotExist(DataModelUtils.GetJobTableName());
            var jobDataContext = new JobDataContext(_storageAccount.TableEndpoint.ToString(),
                                                    _storageAccount.Credentials,
                                                    DataModelUtils.GetJobTableName());
            return jobDataContext.Jobs;
        }

        public double GetFinishedMapPercentage(string jobid, out int totalMapTasks, out int finishedMapCount,
                                               out int pendingMapCount, out int executingMapCount,
                                               out int failedMapCount, out JobDataModel jobData)
        {
            CloudTableClient tableClient = _storageAccount.CreateCloudTableClient();
            string mapTableName = DataModelUtils.GetMapTableName(jobid);
            if (tableClient.DoesTableExist(mapTableName))
            {
                MapMetaDataContext mapContext = new MapMetaDataContext(_storageAccount.TableEndpoint.ToString(),
                                                                       _storageAccount.Credentials, mapTableName);
                JobDataContext jobDataContext = new JobDataContext(_storageAccount.TableEndpoint.ToString(),
                                                                   _storageAccount.Credentials,
                                                                   DataModelUtils.GetJobTableName());

                totalMapTasks = mapContext.MapTasks.ToList().Count;
                JobDataModel job = jobDataContext.Jobs.Where(j => j.RowKey == jobid).First();
                jobData = job;
                //totalMapTasks = job.NumMapTasks;

                if (totalMapTasks != 0)
                {
                    finishedMapCount = totalMapTasks;
                    finishedMapCount =
                        mapContext.MapTasks.Where(
                            p =>
                            ((p.Attempt1Status == MapDataModel.FINISHED) | (p.Attempt2Status == MapDataModel.FINISHED)
                             | (p.Attempt3Status == MapDataModel.FINISHED))).ToList().Count;
                    pendingMapCount =
                        mapContext.MapTasks.Where(p => (p.Attempt1Status == MapDataModel.SUBMITTED)).ToList().Count;

                    executingMapCount =
                        mapContext.MapTasks.Where(
                            p =>
                            ((p.Attempt1Status == MapDataModel.EXECUTING) | (p.Attempt2Status == MapDataModel.EXECUTING) |
                             (p.Attempt3Status == MapDataModel.EXECUTING))).ToList().Count;

                    failedMapCount =
                        mapContext.MapTasks.Where(
                            p =>
                            ((p.Attempt1Status == MapDataModel.FAILED) | (p.Attempt2Status == MapDataModel.FAILED) |
                             (p.Attempt3Status == MapDataModel.FAILED))).ToList().Count;

                    if (job.NumMapTasks > totalMapTasks)
                    {
                        totalMapTasks = job.NumMapTasks;
                    }

                    totalMapTasks = job.NumMapTasks;
                    if (totalMapTasks != 0)
                    {
                        return (finishedMapCount*100)/totalMapTasks;
                    }
                    else
                    {
                        return 0;
                    }
                }
            }
            failedMapCount = 0;
            executingMapCount = 0;
            finishedMapCount = 0;
            pendingMapCount = 0;
            totalMapTasks = 0;
            jobData = null;
            return 0;
        }

        public double GetFinishedReducePercentage(string jobid, out int totalRedTasks, out int finishedRedCount,
                                                  out int pendingRedCount, out int executingRedCount,
                                                  out int failedRedCount)
        {
            CloudTableClient tableClient = _storageAccount.CreateCloudTableClient();
            ReduceMetaDataContext reduceContext = new ReduceMetaDataContext(_storageAccount.TableEndpoint.ToString(),
                                                                            _storageAccount.Credentials,
                                                                            DataModelUtils.GetReduceTableName(jobid));

            totalRedTasks = reduceContext.ReduceTasks.ToList().Count;


            if (totalRedTasks != 0)
            {
                finishedRedCount = totalRedTasks;
                finishedRedCount =
                    reduceContext.ReduceTasks.Where(p => (p.TaskStatus == MapDataModel.FINISHED)).ToList().Count;
                pendingRedCount =
                    reduceContext.ReduceTasks.Where(p => (p.Attempt1Status == MapDataModel.SUBMITTED)).ToList().Count;

                executingRedCount =
                    reduceContext.ReduceTasks.Where(p => (p.TaskStatus == MapDataModel.EXECUTING)).ToList().Count;

                failedRedCount =
                    reduceContext.ReduceTasks.Where(
                        p =>
                        ((p.Attempt1Status == MapDataModel.FAILED) | (p.Attempt2Status == MapDataModel.FAILED) |
                         (p.Attempt3Status == MapDataModel.FAILED))).ToList().Count;


                return finishedRedCount/totalRedTasks*100;
            }

            else
            {
                failedRedCount = 0;
                executingRedCount = 0;
                finishedRedCount = 0;
                pendingRedCount = 0;
                return 0;
            }
        }
    }
}